package org.apache.ocean;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

/**
 *
 * @author jasonr
 */
public class RestartableTimer {
	public static Logger log = Logger.getLogger(RestartableTimer.class.getName());
  private ScheduledExecutorService executor;
  private TimeoutRunnable timeoutRunnable;
  private InternalRunnable internalRunnable;
  private ScheduledFuture future;
  private TimeoutState timeoutState;
  private long delay;
  private long tasktimeout;
  private String name;

  public RestartableTimer(String name, TimeoutRunnable timeoutRunnable, long delay, long tasktimeout) throws Exception {
    this.name = name;
  	this.timeoutRunnable = timeoutRunnable;
    internalRunnable = new InternalRunnable();
    this.delay = delay;
    this.tasktimeout = tasktimeout;
    executor = (ScheduledExecutorService)Executors.newSingleThreadScheduledExecutor();
    runNow();
  }
  
  public void shutdown() throws InterruptedException {
  	log.info("shutdown started "+name);
    cancel(true);
    executor.shutdown();
    log.info("shutdown complete "+name);
  }
  
  public void runNow() throws InterruptedException {
  	//StackTraceElement stackTraceElement = Thread.currentThread().getStackTrace()[3];
  	log.info("runNow isExecuting: "+internalRunnable.isExecuting.get()+" "+name);
    if (internalRunnable.isExecuting.get()) return;
    cancel(true);
    future = executor.scheduleWithFixedDelay(internalRunnable, 0, delay, TimeUnit.MILLISECONDS);
  }
  
  public static interface TimeoutRunnable {
    public void run(TimeoutState timeoutState);
  }
  
  protected class InternalRunnable implements Runnable {
    ReentrantLock lock = new ReentrantLock();
    Condition taskCompletedCondition = lock.newCondition();
    AtomicBoolean isExecuting = new AtomicBoolean(false);

    public InternalRunnable() {
    }

    public void await() throws InterruptedException {
      lock.lock();
      try {
        if (isExecuting.get()) taskCompletedCondition.await();
      } finally {
        lock.unlock();
      }
    }

    public void run() {
      lock.lock();
      isExecuting.set(true);
      try {
        timeoutState = new TimeoutState(tasktimeout);
        timeoutRunnable.run(timeoutState);
        taskCompletedCondition.signalAll();
      } finally {
      	isExecuting.set(false);
        lock.unlock();
      }
    }
  }

  public void cancel(boolean wait) throws InterruptedException {
    if (future != null) {
      future.cancel(false);
      if (timeoutState != null) {
        timeoutState.cancel();
        if (wait) internalRunnable.await();
      }
    }
  }
}
